package de.tuberlin.dima.aim3.oc;

import java.io.IOException;

import de.tuberlin.dima.aim3.oc.util.ParserUtil;
import eu.stratosphere.pact.common.contract.FileDataSink;
import eu.stratosphere.pact.common.contract.FileDataSource;
import eu.stratosphere.pact.common.contract.MapContract;
import eu.stratosphere.pact.common.io.DelimitedInputFormat;
import eu.stratosphere.pact.common.io.FileOutputFormat;
import eu.stratosphere.pact.common.plan.Plan;
import eu.stratosphere.pact.common.plan.PlanAssembler;
import eu.stratosphere.pact.common.plan.PlanAssemblerDescription;
import eu.stratosphere.pact.common.stubs.Collector;
import eu.stratosphere.pact.common.stubs.MapStub;
import eu.stratosphere.pact.common.type.PactRecord;
import eu.stratosphere.pact.common.type.base.PactInteger;
import eu.stratosphere.pact.common.type.base.PactString;

/**
 * PACT plan for the Online Classification project to be used with the
 * stratosphere framework.
 * 
 * @author Florian Feigenbutz <florian.feigenbutz@campus.tu-berlin.de>
 * 
 */
public class WikipediaDumpAnalysis implements PlanAssembler,
    PlanAssemblerDescription {

  @Override
  public String getDescription() {
    return "[noSubTasks] [pathToWikipediaDumpFile] [pathToOutputFile]";
  }

  @Override
  public Plan getPlan(String... args) {

    // parse job parameters
    int noSubTasks = (args.length > 0 ? ParserUtil.parseInt(args[0]) : 1);
    String xmlInputFile = (args.length > 1 ? args[1] : "");
    String outputFile = (args.length > 2 ? args[2] : "");

    // Create DataSourceContract for documents relation
    FileDataSource xmlInput = new FileDataSource(
        JaxbBasedWikipediaDumpInFormat.class, xmlInputFile,
        "Wikipedia Edit History Input");
    xmlInput.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "<page>");
    xmlInput.setParameter(
        DelimiterBasedWikipediaDumpInFormat.CONFIG_KEY_DELIMITER, "<page>");
    xmlInput.setDegreeOfParallelism(noSubTasks);

    MapContract mapper = new MapContract(Mapper.class, "Mapper");
    mapper.setDegreeOfParallelism(noSubTasks);
    mapper.setInput(xmlInput);

    FileDataSink output = new FileDataSink(OutFormat.class, outputFile,
        "Output File");
    output.setDegreeOfParallelism(noSubTasks);
    output.setInput(mapper);

    Plan plan = new Plan(output, "Wikipedia Dump Analysis");
    return plan;
  }

  public static class Mapper extends MapStub {

    @Override
    public void map(PactRecord record, Collector<PactRecord> out)
        throws Exception {
      // TODO Implement proper mapping functionality

      final PactInteger numberOfRevisions = record.getField(0,
          PactInteger.class);
      final PactWikiPage wikiPageValue = record.getField(1, PactWikiPage.class);

      record.setField(0, numberOfRevisions);
      record.setField(1, new PactString(wikiPageValue.toString()));

      out.collect(record);
    }

  }

  public static class OutFormat extends FileOutputFormat {

    private final StringBuilder buffer = new StringBuilder();

    @Override
    public void writeRecord(PactRecord record) throws IOException {
      this.buffer.setLength(0);
      this.buffer.append(record.getField(0, PactInteger.class).getValue());
      this.buffer.append(';');
      this.buffer.append(record.getField(1, PactString.class).toString());
      this.buffer.append('\n');

      byte[] bytes = this.buffer.toString().getBytes();
      this.stream.write(bytes);
    }

  }

}
